package com.st.jms;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;

import com.st.jms.kafka.ProducerClient;
import com.st.jms.util.ReadProperties;

import org.apache.log4j.Logger;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;

public class SendMessageThread extends Thread {
  private static Logger logger = Logger.getLogger(SendMessageThread.class);
  private static Gson gson = new GsonBuilder().setDateFormat("yyyy-MM-dd HH:mm:ss").create();
  private CountDownLatch threadsSignal;
  private Integer roundNum;
  private String queenName;
  private ProducerClient sendMessageClient = new ProducerClient(
          ReadProperties.bundle.getString("brokerList"),100);

  public SendMessageThread(CountDownLatch threadsSignal, Integer roundNum, String queenName) {
    this.threadsSignal = threadsSignal;
    this.roundNum = roundNum;
    this.queenName = queenName;
  }

  public void run() {
    logger.info(Thread.currentThread().getName() + "开始...");
    List<String> messages = new ArrayList<>();
    int k = 0;
    for (int j = 0; j < roundNum; j++) {
      for (int i = 0; i < 1000; i++) {
        Map<String, String> data = new HashMap<>();
        data.put("time", Long.toString(System.currentTimeMillis()));
        data.put("text", String.format("%d", k++));
        data.put("message", "廖永远的简历显示");
        messages.add(gson.toJson(data));
      }
      sendMessageClient.putMessages(queenName, "TestOne", messages);
      messages.clear();
    }
    threadsSignal.countDown();
    logger.info(Thread.currentThread().getName() + "结束。还有" + threadsSignal.getCount() + "个线程");
  }
}
